package com.nami.logic.rocketmq.consumer;

import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.json.JSONUtil;
import com.nami.basic.constants.DownTopicConst;
import com.nami.basic.constants.UpTopicConst;
import com.nami.basic.enums.protocol.DownCMDEnum;
import com.nami.basic.packet.Packet;
import com.nami.basic.request.upmsg.LogoutMqWrap;
import com.nami.basic.bo.mq.DownMsgLogout;
import com.nami.logic.serializer.Serializer;
import com.nami.logic.service.DownMsgSender;
import com.nami.logic.service.PacketFactory;
import com.nami.logic.upmsg.UpMsgLogoutReq;
import com.nami.store.api.request.cache.CacheDelReq;
import com.nami.store.api.service.RemoteL2CacheService;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.dromara.common.core.utils.IdentityUtil;
import org.springframework.stereotype.Component;

import java.util.Date;


@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = UpTopicConst.LOGOUT_TOPIC, consumerGroup = UpTopicConst.LOGOUT_CONSUMER_GROUP,
    consumeMode = ConsumeMode.CONCURRENTLY)
public class LogoutConsumer implements RocketMQListener<String> {

    private final DownMsgSender mqSender;

    private final PacketFactory packetFactory;

    @DubboReference
    private final RemoteL2CacheService cacheService;

    @Override
    public void onMessage(String body) {
        try {
            TimeInterval timer = DateUtil.timer();
            if (StringUtils.isBlank(body)) {
                log.warn("[logout] message is empty");
                return;
            }
            final LogoutMqWrap wrap = JSONUtil.toBean(body, LogoutMqWrap.class);
            if (wrap == null) {
                log.warn("[logout] message is empty");
                return;
            }

            Packet packet = wrap.getPacket();
            final UpMsgLogoutReq request = Serializer.prototype(packet.getSerial()).deserialize(packet.getBody(), UpMsgLogoutReq.class);

            if (request == null
                || StringUtils.isBlank(request.getAppKey())
                || StringUtils.isBlank(request.getUserId())
                || request.getDeviceType() == null
            ) {
                log.warn("[logout] message is empty");
                return;
            }

            if (DateUtil.between(new Date(), new Date(request.getCurrTime()), DateUnit.MINUTE) >= 3) {
                log.warn("[logout] 超时消息不予处理.");
                return;
            }

            CacheDelReq cacheDelReq = new CacheDelReq();
            cacheDelReq.setDeviceType(request.getDeviceType());
            cacheDelReq.setUserId(request.getUserId());
            cacheDelReq.setAppKey(request.getAppKey());
            cacheService.removeOfflineUser(cacheDelReq);

            DownMsgLogout resp = new DownMsgLogout();
            resp.setIdentityNo(IdentityUtil.buildIdentityNo(request.getUserId(), request.getDeviceType()));
            resp.setPacket(packetFactory.newDownmsgSuccPacket(DownCMDEnum.LOGOUT.getCode(), "success"));

            mqSender.syncSendByMq(DownTopicConst.LOGOUT_TOPIC, wrap.getTags(), JSONUtil.toJsonStr(resp));
            log.info("[logout] logout success use:{}ms", timer.interval());
        } catch (Exception var2) {
            var2.printStackTrace();
            log.error("[logout] logout error", var2);
        }

    }


}
